package com.bst.datax.executor.service.jobhandler.streamload;

import com.alibaba.nacos.common.http.param.MediaType;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

//@Service
//@RequiredArgsConstructor
public class DorisStreamLoadService {

    private WebClient dorisWebClient;
    private JdbcTemplate dorisJdbcTemplate;

    //@Retryable(maxAttempts = 3, backoff = @Backoff(delay = 1000))
    //请先了解spring-retry机制
    public void streamLoad(String database, String table, String data, String format) {
        // 构建Stream Load请求
        MultiValueMap<String, String> formData = new LinkedMultiValueMap<>();
        formData.add("format", format);
        formData.add("strip_outer_array", "true");
        formData.add("jsonpaths", "[\"$.id\",\"$.name\",\"$.value\"]");

//        dorisWebClient.post()
//                .uri("/api/" + database + "/" + table + "/_stream_load")
//                .contentType(org.springframework.http.MediaType.valueOf(MediaType.APPLICATION_FORM_URLENCODED))
//                .body(BodyInserters.fromFormData(formData))
//                .bodyValue(data)
//                .retrieve()
//                .bodyToMono(String.class)
//                .doOnSuccess(response -> {
//                    if (response.contains("\"Status\":\"Fail\"")) {
//                        throw new RuntimeException("Stream load failed: " + response);
//                    }
//                })
//                .block();
    }

    public List<Map<String, Object>> extractData(String sql) {
        return dorisJdbcTemplate.queryForList(sql);
    }

    @Scheduled(fixedRate = 60000) // 每分钟执行一次
    public void scheduledDataTransfer() throws JsonProcessingException {
        // 1. 从源表提取数据
        List<Map<String, Object>> data = extractData(
                "SELECT id, name, value FROM source_table WHERE is_processed = 0 LIMIT 10000");

        // 2. 转换为JSON格式
        String jsonData = new ObjectMapper().writeValueAsString(data);

        // 3. 使用Stream Load导入目标表
        streamLoad("your_db", "target_table", jsonData, "json");

        // 4. 标记已处理
        dorisJdbcTemplate.update(
                "UPDATE source_table SET is_processed = 1 WHERE id IN (:ids)",
                new MapSqlParameterSource("ids", data.stream()
                        .map(d -> d.get("id"))
                        .collect(Collectors.toList())));
    }
}
